package ETL

import (
	"context"
	"database/sql"
	"fmt"
	"db2s/go-log/log"
	"db2s/ref"
	"reflect"
	"runtime"
	"strings"
	"time"
)

type WriteOpt struct {
	SqlRows         *sql.Rows
	Schema          string
	Table           string
	TableColumnType []map[string]string
	DB              *sql.DB
	DBType          string
	Tx              *sql.Tx
	StopTime        time.Duration
	SqlLogBin       bool
	SqlMode         bool
	SessionSet      []string
	LobSwitch       bool
	FixRows         int64
	InsertPreFix    string
	Expect          string
	GisValFunc      func(any) any
}

/*
基于预处理的方式进行事务提交
形式：insert into xxx(column1,column2...) value(?,?...)
多用于lob表
*/
func (w WriteOpt) newExecBathInsert(conn *sql.Conn, strSql any) (result FnReturn) {
	var stmt *sql.Stmt
	var execSql [][]*string
	var rowsAffected int64
	var event = "[newExecBathInsert]"
	defer func() {
		if r := ref.RecoverPanic(event, recover()); result.Err == nil && r != nil {
			result.Err = ref.ErrAddPrintf(event, r)
		}
		for _, v2 := range execSql {
			for _, vv2 := range v2 {
				runtime.SetFinalizer(vv2, nil)
			}
		}
		if result.Err = stmt.Close(); result.Err != nil {
			return
		}
	}()
	if stmt, result.Err = conn.PrepareContext(context.Background(), w.InsertPreFix); result.Err != nil {
		result.Err = ref.ErrAddPrintf(event, result.Err)
		return
	}
	switch strSql.(type) {
	case [][]*string:
		execSql = strSql.([][]*string)
	default:
		result.Err = ref.ErrAddPrintf(event, fmt.Errorf(fmt.Sprintf("not metch type. curry type %v", reflect.TypeOf(strSql).String())))
		return
	}
	for _, v := range execSql {
		args := make([]interface{}, strings.Count(w.InsertPreFix, "?"))
		j := 0 // 新的索引变量
		for _, v3 := range v {
			if strings.EqualFold(*v3, "<prepare nil>") {
				args[j] = nil
				j++
			} else if strings.EqualFold(*v3, "<prepare entry>") { //处理text类型的空字符串
				args[j] = reflect.ValueOf("").Interface()
				j++
			} else if strings.EqualFold(*v3, "<null>") {
				args[j] = sql.NullString{Valid: false}
				j++
			} else if strings.Contains(*v3, ")SRID:") {
				q := strings.Split(*v3, "SRID:")
				args[j] = q[0]
				j++
				args[j] = q[1]
				j++
			} else {
				args[j] = reflect.ValueOf(*v3).Interface()
				j++
			}
		}
		var rowsAffect sql.Result
		if rowsAffect, result.Err = stmt.Exec(args...); result.Err != nil {
			return
		}
		var pp int64
		if pp, result.Err = rowsAffect.RowsAffected(); result.Err != nil {
			result.Err = ref.ErrAddPrintf(event, result.Err)
			return
		}
		rowsAffected += pp
	}
	result.Result = rowsAffected
	return
}
func (w WriteOpt) insertExec(conn *sql.Conn, setVariables any) (result FnReturn) {
	var rowsAffected int64
	var execSql []*string
	var event = "[insertExec]"
	defer func() {
		if r := ref.RecoverPanic(event, recover()); result.Err == nil && r != nil {
			result.Err = ref.ErrAddPrintf(event, r)
		}
		result.Result = rowsAffected
		for _, v := range execSql {
			runtime.SetFinalizer(v, nil)
		}
	}()
	for _, v := range getBaseExecSql(setVariables) {
		var q int64
		if q, result.Err = baseExecSql(conn, v); result.Err != nil {
			result.Err = ref.ErrAddPrintf(event, result.Err)
			return
		}
		rowsAffected += q
	}
	return
}

// NormalInsert 开启一个参数 /*
func (w WriteOpt) NormalInsert(sessionOptions []string, setVariables any) (result FnReturn) {
	var (
		ctx    context.Context
		cancel context.CancelFunc
		conn   *sql.Conn
		event  = "[NormalInsert]"
	)
	if w.StopTime == 0 {
		ctx = context.Background()
	} else {
		ctx, cancel = context.WithTimeout(context.Background(), w.StopTime)
	}
	beginTime := time.Now()
	defer func() {
		if r := ref.RecoverPanic(event, recover()); result.Err == nil && r != nil {
			result.Err = ref.ErrAddPrintf(event, r)
		}
		if cancel != nil {cancel()}
		result.ExecTime = time.Since(beginTime).Milliseconds()
	}()
	// 从连接池获取连接
	if conn, result.Err = w.DB.Conn(ctx); result.Err != nil {
		result.Err = ref.ErrAddPrintf(event, result.Err)
	}
	defer conn.Close()
	if result.Err = setConnectionSessionVariables(conn, sessionOptions); result.Err != nil {
		result.Err = ref.ErrAddPrintf(event, result.Err)
		return
	}
	if w.LobSwitch {
		return w.newExecBathInsert(conn, setVariables)
	}
	return w.insertExec(conn, setVariables)
}

/*
基于预处理的方式进行事务提交
形式：insert into xxx(column1,column2...) value(?,?...)
多用于lob表
*/
func (w WriteOpt) trxExecBathInsert(_ context.Context, trx *sql.Tx, strSql any) (err error) {
	var event = "[trxExecBathInsert]"
	defer func() {
		if r := ref.RecoverPanic(event, recover()); err == nil && r != nil {
			err = ref.ErrAddPrintf(event, r)
		}
	}()
	insertSqlPerf := w.InsertPreFix
	strSqlExec := strSql
	stmt, err := trx.Prepare(insertSqlPerf)
	if err != nil {
		err = fmt.Errorf(fmt.Sprintf("trx prepare fail. error is %v", err))
		panic(err.Error())
		return err
	}
	defer func() {
		err1 := stmt.Close()
		if err1 != nil {
			return
		}
	}()
	switch fmt.Sprintf("%v", reflect.TypeOf(strSqlExec)) {
	case GarthStringPointGarth:
		for _, v := range strSqlExec.([][]*string) {
			args := make([]interface{}, len(v))
			for i, v3 := range v {
				if strings.EqualFold(*v3, "<prepare nil>") {
					args[i] = nil
				} else if strings.EqualFold(*v3, "<prepare entry>") { //处理text类型的空字符串
					args[i] = reflect.ValueOf("").Interface()
				} else if strings.EqualFold(*v3, "<null>") {
					args[i] = sql.NullString{Valid: false}
				} else {
					args[i] = reflect.ValueOf(v3).Interface()
				}
			}
			_, err = trx.Stmt(stmt).Exec(args...)
			if err != nil {
				err = fmt.Errorf(fmt.Sprintf("trx exec fail. error is %v", err))
				// 处理插入操作错误
				if err1 := trx.Rollback(); err1 != nil {
					err = fmt.Errorf(fmt.Sprintf("%v. trx rollback fail. error is %v", err, err1))
				}
				return err
			}
			for _, v2 := range v {
				runtime.SetFinalizer(v2, nil)
			}
		}
	}
	return nil
}
func (w WriteOpt) BeginTrxInsert(sessionOptions []string, setVariables any) (result FnReturn) {
	//var (
	//	trx    *sql.Tx
	//	event  = "[BeginTrxInsert]"
	//	ctx    context.Context
	//	conn   *sql.Conn
	//	cancel context.CancelFunc
	//)
	//defer func() {
	//	if r := ref.RecoverPanic(event, recover()); result.Err == nil && r != nil {
	//		result.Err = ref.ErrAddPrintf(event, r)
	//	}
	//}()
	//beginTime := time.Now()
	//if result.Err = setConnectionSessionVariables(w.DB, sessionOptions); result.Err != nil {
	//	result.Err = ref.ErrAddPrintf(event, result.Err)
	//	return
	//}
	//if w.StopTime > 0 {
	//	ctx, cancel = context.WithTimeout(context.Background(), w.StopTime)
	//	defer cancel()
	//	if trx, result.Err = w.DB.BeginTx(ctx, &sql.TxOptions{
	//		Isolation: sql.LevelReadCommitted, // 设置事务隔离级别
	//		ReadOnly:  false,                  // 设置事务为读写模式
	//	}); result.Err != nil {
	//		result.Err = fmt.Errorf(fmt.Sprintf("context_withTimeout failed to open transaction,context timeout is {%v}. error is {%v}", w.StopTime, result.Err))
	//		return
	//	}
	//} else {
	//	if trx, result.Err = beginTrx(w.DB); result.Err != nil {
	//		result.Err = fmt.Errorf(fmt.Sprintf("failed to open transaction. error is {%v}", result.Err))
	//		return
	//	}
	//}
	//if _, err := w.DB.Exec(fmt.Sprintf("select @@transaction_isolation")); err != nil {
	//	return FnReturn{}
	//}
	//for _, v := range w.SessionSet {
	//	if result.Err = setSessionVariables(ctx, trx, v); result.Err != nil {
	//		return
	//	}
	//}
	//if w.LobSwitch {
	//	if result.Err = w.trxExecBathInsert(ctx, trx, setVariables); result.Err != nil {
	//		return
	//	}
	//} else {
	//	switch fmt.Sprintf("%v", reflect.TypeOf(setVariables)) {
	//	case StringPoint:
	//		if result.Err = setSessionVariables(ctx, trx, *setVariables.(*string)); result.Err != nil {
	//			result.Err = fmt.Errorf(fmt.Sprintf("exec sql is %v error info is %v", *setVariables.(*string), result.Err))
	//			return
	//		}
	//		runtime.SetFinalizer(setVariables, nil)
	//	case StringPointGarth:
	//		for _, v := range setVariables.([]*string) {
	//			if result.Err = setSessionVariables(ctx, trx, *v); result.Err != nil {
	//				result.Err = fmt.Errorf(fmt.Sprintf("exec sql is %v error info is %v", *v, result.Err))
	//				return
	//			}
	//			runtime.SetFinalizer(v, nil)
	//		}
	//	}
	//}
	//if result.Err = trxCommit(trx); result.Err != nil {
	//	return
	//}
	//result.ExecTime = time.Since(beginTime).Milliseconds()
	return
}

// DDLUpdate 开启一个参数 /*
func (w WriteOpt) DDLUpdate(db *sql.DB, sessionOptions []string, setVariables any) (result FnReturn) {
	var (
		ctx    context.Context
		event  = "[DDLUpdate]"
		conn   *sql.Conn
		cancel context.CancelFunc
	)
	defer func() {
		if r := ref.RecoverPanic(event, recover()); result.Err == nil && r != nil {
			result.Err = ref.ErrAddPrintf(event, r)
		}
	}()
	if w.StopTime == 0 {
		ctx = context.Background()
	} else {
		ctx, cancel = context.WithTimeout(context.Background(), w.StopTime)
	}
	if cancel != nil {
		defer cancel()
	}
	beginTime := time.Now()
	// 从连接池获取连接
	if conn, result.Err = db.Conn(ctx); result.Err != nil {
		log.ErrorLog().Error(event, db, result.Err)
	}
	defer conn.Close()
	if result.Err = setConnectionSessionVariables(conn, sessionOptions); result.Err != nil {
		result.Err = ref.ErrAddPrintf(event, result.Err)
		return
	}
	for _, v := range getBaseExecSql(setVariables) {
		if _, result.Err = baseExecSql(conn, v); result.Err != nil {
			result.Err = ref.ErrAddPrintf(event, result.Err)
			return
		}
	}
	result.ExecTime = time.Since(beginTime).Milliseconds()
	return
}

// normalUpdate 开启一个参数 /*
func (w WriteOpt) normalUpdate(sessionOptions []string, setVariables any) (result FnReturn) {
	var event = "[normalUpdate]"
	var conn *sql.Conn
	defer func() {
		if r := ref.RecoverPanic(event, recover()); result.Err == nil && r != nil {
			result.Err = ref.ErrAddPrintf(event, r)
		}
	}()
	ctx, cancel := context.WithTimeout(context.Background(), w.StopTime)
	defer cancel()
	beginTime := time.Now()
	// 从连接池获取连接
	if conn, result.Err = w.DB.Conn(ctx); result.Err != nil {
		return
	}
	defer conn.Close()
	if result.Err = setConnectionSessionVariables(conn, sessionOptions); result.Err != nil {
		result.Err = ref.ErrAddPrintf(event, result.Err)
		return
	}
	for _, v := range getBaseExecSql(setVariables) {
		if _, result.Err = baseExecSql(conn, v); result.Err != nil {
			result.Err = ref.ErrAddPrintf(event, result.Err)
			return
		}
	}
	result.ExecTime = time.Since(beginTime).Milliseconds()
	return
}

// BeginTrxUpdate 开启一个参数 /*
func (w WriteOpt) BeginTrxUpdate(sessionOptions []string, setVariables any) (result FnReturn) {
	//var (
	//	trx    *sql.Tx
	//	ctx    context.Context
	//	cancel context.CancelFunc
	//	event  = "[BeginTrxUpdate]"
	//	conn   *sql.Conn
	//)
	//defer func() {
	//	if r := ref.RecoverPanic(event, recover()); result.Err == nil && r != nil {
	//		result.Err = r
	//	}
	//}()
	//beginTime := time.Now()
	//if result.Err = setConnectionSessionVariables(w.DB, sessionOptions); result.Err != nil {
	//	result.Err = ref.ErrAddPrintf(event, result.Err)
	//	return
	//}
	//if w.StopTime > 0 {
	//	ctx, cancel = context.WithTimeout(context.Background(), w.StopTime)
	//	defer cancel()
	//	if trx, result.Err = w.DB.BeginTx(ctx, &sql.TxOptions{
	//		Isolation: sql.LevelReadCommitted, // 设置事务隔离级别
	//		ReadOnly:  false,                  // 设置事务为读写模式
	//	}); result.Err != nil {
	//		result.Err = fmt.Errorf(fmt.Sprintf("context_withTimeout failed to open transaction,context timeout is {%v}. error is {%v}", w.StopTime, result.Err))
	//		return
	//	}
	//} else {
	//	if trx, result.Err = beginTrx(w.DB); result.Err != nil {
	//		result.Err = fmt.Errorf(fmt.Sprintf("failed to open transaction. error is {%v}", result.Err))
	//		return
	//	}
	//}
	//if len(w.SessionSet) > 0 {
	//	for _, v := range w.SessionSet {
	//		if result.Err = setSessionVariables(ctx, trx, v); result.Err != nil {
	//			return
	//		}
	//	}
	//}
	//switch fmt.Sprintf("%v", reflect.TypeOf(setVariables)) {
	//case StringPoint:
	//	if result.Err = setSessionVariables(ctx, trx, *setVariables.(*string)); result.Err != nil {
	//		result.Err = fmt.Errorf(fmt.Sprintf("exec sql is %v error info is %v", *setVariables.(*string), result.Err))
	//		return
	//	}
	//	runtime.SetFinalizer(setVariables, nil)
	//case StringPointGarth:
	//	for _, v := range setVariables.([]*string) {
	//		if result.Err = setSessionVariables(ctx, trx, *v); result.Err != nil {
	//			result.Err = fmt.Errorf(fmt.Sprintf("exec sql is %v error info is %v", *v, result.Err))
	//			return
	//		}
	//		runtime.SetFinalizer(v, nil)
	//	}
	//}
	//if result.Err = trxCommit(trx); result.Err != nil {
	//	return
	//}
	//result.ExecTime = time.Since(beginTime).Milliseconds()
	return
}

func Insert(w any, sessionOptions []string, setVariables any) (result FnReturn) {
	defer func() {
		if r := ref.RecoverPanic("[Insert]", recover()); result.Err == nil && r != nil {
			result.Err = r
		}
	}()
	switch w.(type) {
	case WriteOpt:
		result = retriesExecSql(func(sessionOptions []string, setVariables any) FnReturn {
			return w.(WriteOpt).NormalInsert(sessionOptions, setVariables)
		}, sessionOptions, setVariables)
		return
	}
	return
}
func TrxInsert(w any, sessionOptions []string, setVariables any) (execTime int64, err error) {
	defer func() {
		if r := ref.RecoverPanic("[TrxInsert]", recover()); err == nil && r != nil {
			err = r
		}
	}()
	switch w.(type) {
	case WriteOpt:
		p := retriesExecSql(func(sessionOptions []string, setVariables any) FnReturn {
			return w.(WriteOpt).BeginTrxInsert(sessionOptions, setVariables)
		}, sessionOptions, setVariables)
		return p.ExecTime, p.Err
	}
	return
}
func Update(w any, sessionOptions []string, setVariables any) (execTime int64, err error) {
	defer func() {
		if r := ref.RecoverPanic("[Update]", recover()); err == nil && r != nil {
			err = r
		}
	}()
	switch w.(type) {
	case WriteOpt:
		p := retriesExecSql(func(sessionOptions []string, setVariables any) FnReturn {
			return w.(WriteOpt).normalUpdate(sessionOptions, setVariables)
		}, sessionOptions, setVariables)
		return p.ExecTime, p.Err
	}
	return
}
func TrxUpdate(w any, sessionOptions []string, setVariables any) (execTime int64, err error) {
	defer func() {
		if r := ref.RecoverPanic("[TrxUpdate]", recover()); err == nil && r != nil {
			err = r
		}
	}()
	switch w.(type) {
	case WriteOpt:
		p := retriesExecSql(func(sessionOptions []string, setVariables any) FnReturn {
			return w.(WriteOpt).BeginTrxUpdate(sessionOptions, setVariables)
		}, sessionOptions, setVariables)
		return p.ExecTime, p.Err
	}
	return
}
func DDL(w any, db *sql.DB, sessionOptions []string, setVariables any) (execTime int64, err error) {
	defer func() {
		if r := ref.RecoverPanic("[DDL]", recover()); err == nil && r != nil {
			err = r
		}
	}()
	switch w.(type) {
	case WriteOpt:
		p := retriesExecSql(func(sessionOptions []string, setVariables any) FnReturn {
			return w.(WriteOpt).DDLUpdate(db, sessionOptions, setVariables)
		}, sessionOptions, setVariables)
		return p.ExecTime, p.Err
	}
	return
}
